package com.ihr360.job.core.repository.dao;

import com.ihr360.job.core.BatchStatus;
import com.ihr360.job.core.ExitStatus;
import com.ihr360.job.core.entity.JobExecution;
import com.ihr360.job.core.entity.StepExecution;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

/**
 * JDBC implementation of {@link StepExecutionDao}.<br>
 * <p>
 * Allows customization of the tables names used by Spring Batch for step meta
 * data via a prefix property.<br>
 * <p>
 * Uses sequences or tables (via Spring's {@link DataFieldMaxValueIncrementer}
 * abstraction) to create all primary keys before inserting a new row. All
 * objects are checked to ensure all fields to be stored are not null. If any
 * are found to be null, an IllegalArgumentException will be thrown. This could
 * be left to JdbcTemplate, however, the exception will be fairly vague, and
 * fails to highlight which field caused the exception.<br>
 *
 * @author Lucas Ward
 * @author Dave Syer
 * @author Robert Kasanicky
 * @author David Turanski
 * @see StepExecutionDao
 */
public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implements StepExecutionDao, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(JdbcStepExecutionDao.class);

    private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_NAME, JOB_EXECUTION_ID, START_TIME, "
            + "END_TIME, STATUS, COMMIT_COUNT, READ_COUNT, FILTER_COUNT, WRITE_COUNT, EXIT_CODE, EXIT_DESC,EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, PROCESS_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED) "
            + "values(?,?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

    private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, "
            + "STATUS = ?, COMMIT_COUNT = ?, READ_COUNT = ?, FILTER_COUNT = ?, WRITE_COUNT = ?, EXIT_CODE = ?, "
            + "EXIT_DESC=?,EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, PROCESS_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ?, ROLLBACK_COUNT = ?, LAST_UPDATED = ?"
            + " where STEP_EXECUTION_ID = ? and VERSION = ?";

    private static final String GET_RAW_STEP_EXECUTIONS = "SELECT STEP_EXECUTION_ID, STEP_NAME, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
            + " READ_COUNT, FILTER_COUNT, WRITE_COUNT, EXIT_CODE,EXIT_DESC, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, PROCESS_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED, VERSION from %PREFIX%STEP_EXECUTION where JOB_EXECUTION_ID = ?";

    private static final String GET_STEP_EXECUTIONS = GET_RAW_STEP_EXECUTIONS + " order by STEP_EXECUTION_ID";

    private static final String GET_STEP_EXECUTION = GET_RAW_STEP_EXECUTIONS + " and STEP_EXECUTION_ID = ?";

    private static final String CURRENT_VERSION_STEP_EXECUTION = "SELECT VERSION FROM %PREFIX%STEP_EXECUTION WHERE STEP_EXECUTION_ID=?";

    private int exitMessageLength = DEFAULT_EXIT_MESSAGE_LENGTH;

    private DataFieldMaxValueIncrementer stepExecutionIncrementer;

    /**
     * Public setter for the exit message length in database. Do not set this if
     * you haven't modified the schema.
     *
     * @param exitMessageLength the exitMessageLength to set
     */
    public void setExitMessageLength(int exitMessageLength) {
        this.exitMessageLength = exitMessageLength;
    }

    public void setStepExecutionIncrementer(DataFieldMaxValueIncrementer stepExecutionIncrementer) {
        this.stepExecutionIncrementer = stepExecutionIncrementer;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet();
        Assert.notNull(stepExecutionIncrementer, "StepExecutionIncrementer cannot be null.");
    }

    /**
     * Save a StepExecution. A unique id will be generated by the
     * stepExecutionIncrementor, and then set in the StepExecution. All values
     * will then be stored via an INSERT statement.
     *
     * @see StepExecutionDao#saveStepExecution(StepExecution)
     */
    @Override
    public void saveStepExecution(StepExecution stepExecution) {
        List<Object[]> parameters = buildStepExecutionParameters(stepExecution);
        Object[] parameterValues = parameters.get(0);

        //Template expects an int array fails with Integer
        int[] parameterTypes = new int[parameters.get(1).length];
        for (int i = 0; i < parameterTypes.length; i++) {
            parameterTypes[i] = (Integer) parameters.get(1)[i];
        }

        getJdbcTemplate().update(getQuery(SAVE_STEP_EXECUTION), parameterValues, parameterTypes);
    }

    /**
     * Batch insert StepExecutions
     *
     * @see StepExecutionDao#saveStepExecutions(Collection)
     */
    @Override
    public void saveStepExecutions(final Collection<StepExecution> stepExecutions) {
        Assert.notNull(stepExecutions, "Attempt to save a null collection of step executions");

        if (!stepExecutions.isEmpty()) {
            final Iterator<StepExecution> iterator = stepExecutions.iterator();
            getJdbcTemplate().batchUpdate(getQuery(SAVE_STEP_EXECUTION), new BatchPreparedStatementSetter() {

                @Override
                public int getBatchSize() {
                    return stepExecutions.size();
                }

                @Override
                public void setValues(PreparedStatement ps, int i) throws SQLException {
                    StepExecution stepExecution = iterator.next();
                    List<Object[]> parameters = buildStepExecutionParameters(stepExecution);
                    Object[] parameterValues = parameters.get(0);
                    Integer[] parameterTypes = (Integer[]) parameters.get(1);
                    for (int indx = 0; indx < parameterValues.length; indx++) {
                        switch (parameterTypes[indx]) {
                            case Types.INTEGER:
                                ps.setInt(indx + 1, (Integer) parameterValues[indx]);
                                break;
                            case Types.VARCHAR:
                                ps.setString(indx + 1, (String) parameterValues[indx]);
                                break;
                            case Types.TIMESTAMP:
                                if (parameterValues[indx] != null) {
                                    ps.setTimestamp(indx + 1, new Timestamp(((java.util.Date) parameterValues[indx]).getTime()));
                                } else {
                                    ps.setNull(indx + 1, Types.TIMESTAMP);
                                }
                                break;
                            case Types.BIGINT:
                                ps.setLong(indx + 1, (Long) parameterValues[indx]);
                                break;
                            default:
                                throw new IllegalArgumentException(
                                        "unsupported SQL parameter type for step execution field index " + i);
                        }
                    }
                }
            });
        }
    }

    private List<Object[]> buildStepExecutionParameters(StepExecution stepExecution) {
        Assert.isNull(stepExecution.getId(),
                "to-be-saved (not updated) StepExecution can't already have an id assigned");
        Assert.isNull(stepExecution.getVersion(),
                "to-be-saved (not updated) StepExecution can't already have a version assigned");
        validateStepExecution(stepExecution);
        stepExecution.setId(stepExecutionIncrementer.nextLongValue());
        stepExecution.incrementVersion(); //Should be 0
        List<Object[]> parameters = new ArrayList<Object[]>();
        String exitDescription = truncateExitDescription(stepExecution.getExitStatus().getExitDescription());
        String exitMessage = truncateExitDescription(stepExecution.getExitStatus().getExitMessage());
        Object[] parameterValues = new Object[]{stepExecution.getId(), stepExecution.getVersion(),
                stepExecution.getStepName(), stepExecution.getJobExecutionId(), stepExecution.getStartTime(),
                stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(),
                stepExecution.getReadCount(), stepExecution.getFilterCount(), stepExecution.getWriteCount(),
                stepExecution.getExitStatus().getExitCode(), exitDescription,exitMessage, stepExecution.getReadSkipCount(),
                stepExecution.getWriteSkipCount(), stepExecution.getProcessSkipCount(),
                stepExecution.getRollbackCount(), stepExecution.getLastUpdated()};
        Integer[] parameterTypes = new Integer[]{Types.BIGINT, Types.INTEGER, Types.VARCHAR, Types.BIGINT,
                Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER,
                Types.INTEGER, Types.VARCHAR, Types.VARCHAR,Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER,
                Types.INTEGER, Types.TIMESTAMP};
        parameters.add(0, Arrays.copyOf(parameterValues, parameterValues.length));
        parameters.add(1, Arrays.copyOf(parameterTypes, parameterTypes.length));
        return parameters;
    }

    /**
     * Validate StepExecution. At a minimum, JobId, StartTime, and Status cannot
     * be null. EndTime can be null for an unfinished job.
     *
     * @throws IllegalArgumentException
     */
    private void validateStepExecution(StepExecution stepExecution) {
        Assert.notNull(stepExecution);
        Assert.notNull(stepExecution.getStepName(), "StepExecution step name cannot be null.");
        Assert.notNull(stepExecution.getStartTime(), "StepExecution start time cannot be null.");
        Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null.");
    }

    @Override
    public void updateStepExecution(StepExecution stepExecution) {

        validateStepExecution(stepExecution);
        Assert.notNull(stepExecution.getId(), "StepExecution Id cannot be null. StepExecution must saved"
                + " before it can be updated.");

        // Do not check for existence of step execution considering
        // it is saved at every commit point.

        String exitDescription = truncateExitDescription(stepExecution.getExitStatus().getExitDescription());
        String exitMessage = truncateExitDescription(stepExecution.getExitStatus().getExitMessage());
        // Attempt to prevent concurrent modification errors by blocking here if
        // someone is already trying to do it.
        synchronized (stepExecution) {

            Integer version = stepExecution.getVersion() + 1;
            Object[] parameters = new Object[]{stepExecution.getStartTime(), stepExecution.getEndTime(),
                    stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getReadCount(),
                    stepExecution.getFilterCount(), stepExecution.getWriteCount(),
                    stepExecution.getExitStatus().getExitCode(), exitDescription,exitMessage, version,
                    stepExecution.getReadSkipCount(), stepExecution.getProcessSkipCount(),
                    stepExecution.getWriteSkipCount(), stepExecution.getRollbackCount(),
                    stepExecution.getLastUpdated(), stepExecution.getId(), stepExecution.getVersion()};
            int count = getJdbcTemplate()
                    .update(getQuery(UPDATE_STEP_EXECUTION),
                            parameters,
                            new int[]{Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER,
                                    Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.VARCHAR,Types.VARCHAR, Types.INTEGER,
                                    Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.TIMESTAMP,
                                    Types.BIGINT, Types.INTEGER});

            // Avoid concurrent modifications...
            if (count == 0) {
                int curentVersion = getJdbcTemplate().queryForObject(getQuery(CURRENT_VERSION_STEP_EXECUTION),
                        new Object[]{stepExecution.getId()}, Integer.class);
                throw new OptimisticLockingFailureException("Attempt to update step execution id="
                        + stepExecution.getId() + " with wrong version (" + stepExecution.getVersion()
                        + "), where current version is " + curentVersion);
            }

            stepExecution.incrementVersion();

        }
    }

    /**
     * Truncate the exit description if the length exceeds
     * {@link #DEFAULT_EXIT_MESSAGE_LENGTH}.
     *
     * @param description the string to truncate
     * @return truncated description
     */
    private String truncateExitDescription(String description) {
        if (description != null && description.length() > exitMessageLength) {
            if (logger.isDebugEnabled()) {
                logger.debug("Truncating long message before update of StepExecution, original message is: " + description);
            }
            return description.substring(0, exitMessageLength);
        } else {
            return description;
        }
    }

    @Override
    public StepExecution getStepExecution(JobExecution jobExecution, Long stepExecutionId) {
        List<StepExecution> executions = getJdbcTemplate().query(getQuery(GET_STEP_EXECUTION),
                new StepExecutionRowMapper(jobExecution), jobExecution.getId(), stepExecutionId);

        Assert.state(executions.size() <= 1,
                "There can be at most one step execution with given name for single job execution");
        if (executions.isEmpty()) {
            return null;
        } else {
            return executions.get(0);
        }
    }

    @Override
    public List<StepExecution> getStepExecutions(JobExecution jobExecution) {
        if (jobExecution==null){
            return Collections.emptyList();
        }
        List<StepExecution> executions = getJdbcTemplate().query(getQuery(GET_STEP_EXECUTIONS),
                new StepExecutionRowMapper(jobExecution), jobExecution.getId());
        if (CollectionUtils.isEmpty(executions)) {
            return Collections.emptyList();
        }
        return executions;
    }

    @Override
    public void addStepExecutions(JobExecution jobExecution) {
        getJdbcTemplate().query(getQuery(GET_STEP_EXECUTIONS), new StepExecutionRowMapper(jobExecution),
                jobExecution.getId());
    }

    private static class StepExecutionRowMapper implements RowMapper<StepExecution> {

        private final JobExecution jobExecution;

        public StepExecutionRowMapper(JobExecution jobExecution) {
            this.jobExecution = jobExecution;
        }

        @Override
        public StepExecution mapRow(ResultSet rs, int rowNum) throws SQLException {
            StepExecution stepExecution = new StepExecution(rs.getString(2), jobExecution, rs.getLong(1));
            stepExecution.setStartTime(rs.getTimestamp(3));
            stepExecution.setEndTime(rs.getTimestamp(4));
            stepExecution.setStatus(BatchStatus.valueOf(rs.getString(5)));
            stepExecution.setCommitCount(rs.getInt(6));
            stepExecution.setReadCount(rs.getInt(7));
            stepExecution.setFilterCount(rs.getInt(8));
            stepExecution.setWriteCount(rs.getInt(9));
            stepExecution.setExitStatus(new ExitStatus(rs.getString(10), rs.getString(11),rs.getString(12)));
            stepExecution.setReadSkipCount(rs.getInt(13));
            stepExecution.setWriteSkipCount(rs.getInt(14));
            stepExecution.setProcessSkipCount(rs.getInt(15));
            stepExecution.setRollbackCount(rs.getInt(16));
            stepExecution.setLastUpdated(rs.getTimestamp(17));
            stepExecution.setVersion(rs.getInt(18));
            return stepExecution;
        }

    }

}
